package abacab.ext;

import java.io.IOException;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.Collections;
import java.util.Enumeration;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import abacab.ActorSystem;
import abacab.ref.ActorLocator;
import abacab.ref.ActorRef;
import abacab.remote.ConnectionRegistry;

import com.typesafe.config.Config;
import com.typesafe.config.ConfigException;
import com.typesafe.config.ConfigException.Generic;
import com.typesafe.config.ConfigException.IO;
import com.typesafe.config.ConfigException.Missing;
import com.typesafe.config.ConfigException.WrongType;

/**
 * The extension for remote
 */
public class RemotingExtension implements Extension {
	/** The logger */
	private static final Logger																	logger						= LoggerFactory
																																										.getLogger(RemotingExtension.class);

	/** The remote configuration key */
	private static final String																	REMOTE_CONFIG			= "abacab.remote";

	/** The remote interface configuration key */
	private static final String																	INTERFACE_CONFIG	= "interface";

	/** The remote port configuration key */
	private static final String																	PORT_CONFIG				= "port";

	/** The protocol for remote actors */
	public static final String																	PROTOCOL					= "abba://";

	/** The default listening port */
	public static final int																			DEFAULT_PORT			= 5432;

	/** The locator */
	@SuppressWarnings("unused")
	private static final RemoteActorLocator											LOCATOR						= new RemoteActorLocator();

	/** The connection registers (per interface & port) */
	private static final ConcurrentMap<String, RegistryWrapper>	registry					= new ConcurrentHashMap<String, RegistryWrapper>();

	/** The connection registry */
	private final RegistryWrapper																connectionRegistry;

	/**
	 * Create the extension
	 * 
	 * @param actorSystem
	 *          the actor system
	 * @throws ConfigException
	 *           if the configuration is invalid
	 */
	public RemotingExtension(ActorSystem actorSystem) throws ConfigException {
		Config remoteConfig = actorSystem.getConfig().getConfig(REMOTE_CONFIG);
		try {
			InetAddress netInterface = getNetInterface(remoteConfig);
			int port = remoteConfig.getInt(PORT_CONFIG);
			String key = netInterface + ":" + port;
			RegistryWrapper wrapper = registry.get(key);
			if (wrapper == null) {
				wrapper = new RegistryWrapper(new ConnectionRegistry(actorSystem, netInterface, port));
				RegistryWrapper concWrapper = registry.putIfAbsent(key, wrapper);
				if (concWrapper != null) {
					wrapper = concWrapper;
					wrapper.addSystem();
				}
				else {
					wrapper.getRegistry().init();
				}
			}
			this.connectionRegistry = wrapper;
		}
		catch (IOException e) {
			throw new IO(remoteConfig.origin(), e.getMessage(), e);
		}
	}

	/**
	 * Get the network interface
	 * 
	 * @param remoteConfig
	 *          the configuration
	 * @return the network interface
	 * @throws WrongType
	 *           if the configuration is invalid
	 * @throws UnknownHostException
	 *           if the host interface is unknown
	 * @throws SocketException
	 *           if the search for a suitable interface failed (none was provided in the configuration)
	 */
	private final InetAddress getNetInterface(Config remoteConfig) throws WrongType, UnknownHostException,
			SocketException {
		try {
			return InetAddress.getByName(remoteConfig.getString(INTERFACE_CONFIG));
		}
		catch (Missing e) {
			// The parameter is missing, ignore the exception and try to find one suitable
		}
		Enumeration<NetworkInterface> nets = NetworkInterface.getNetworkInterfaces();
		for (NetworkInterface netint : Collections.list(nets)) {
			if (!netint.isLoopback() && netint.isUp()) {
				Enumeration<InetAddress> inetAddresses = netint.getInetAddresses();
				while (inetAddresses.hasMoreElements()) {
					InetAddress addr = inetAddresses.nextElement();
					if (addr instanceof Inet4Address) {
						return addr;
					}
				}
			}
		}
		throw new Generic("No interface was specified in the configuration & no suitable interface was found");
	}

	@Override
	public void dispose() {
		if (this.connectionRegistry.removeSystem() == 0) {
			ConnectionRegistry cr = this.connectionRegistry.getRegistry();
			registry.remove(cr.getNetworkInterface() + ":" + cr.getListeningPort());
			cr.stop();
		}
	}

	/**
	 * Wraps a registry
	 */
	private static final class RegistryWrapper {
		/** The connection registry */
		private final ConnectionRegistry	wrapedRegistry;

		/** The number of systems using it */
		private final AtomicInteger				systems;

		/**
		 * Create the wrapper
		 * 
		 * @param registry
		 *          the connection registry
		 */
		RegistryWrapper(ConnectionRegistry registry) {
			this.wrapedRegistry = registry;
			this.systems = new AtomicInteger(1);
		}

		/**
		 * @return the connection registry
		 */
		ConnectionRegistry getRegistry() {
			return this.wrapedRegistry;
		}

		/**
		 * Add a system
		 */
		void addSystem() {
			this.systems.incrementAndGet();
		}

		/**
		 * Remove a system
		 * 
		 * @return the number of remaining systems using it
		 */
		int removeSystem() {
			return this.systems.decrementAndGet();
		}
	}

	/**
	 * The actor locator for remoting
	 */
	private static final class RemoteActorLocator implements ActorLocator {
		/**
		 * Create the locator
		 */
		RemoteActorLocator() {
			ActorSystem.registerLocator(this);
		}

		@Override
		@SuppressWarnings({ "synthetic-access", "unchecked" })
		public <M> ActorRef<M> locateActor(String path) throws IllegalArgumentException {
			try {
				for (RegistryWrapper wrapper : registry.values()) {
					try {
						ActorRef<M> actorRef = wrapper.getRegistry().getRemoteActor(path);
						if (actorRef != null) {
							return actorRef;
						}
					}
					catch (IOException e) {
						logger.warn("Failed to interact with the remote actor system: " + path);
					}
				}
			}
			catch (InterruptedException e) {
				Thread.currentThread().interrupt();
				logger.warn("Failed to get the remote actor: the thread was interrupted before getting the result");
			}
			return null;
		}
	}
}